package com.lpssfxy.kafka;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

public class LogProcessor implements Processor<byte[], byte[]> {
    // Initialize this processor with the given context.
    private ProcessorContext processorContext;

    @Override
    public void init(ProcessorContext context) {
        this.processorContext = context;
    }

    // Process the record with the given key and value.
    @Override
    public void process(byte[] key, byte[] value) {
        // 核心处理流程
        String input = new String(value);
        // 提取数据，以固定前缀过滤日志信息
        if (input.contains("PRODUCT_RATING_PREFIX:")) {
            //System.out.println("product rating log coming >>>>>> " + input);
            input = input.split("PRODUCT_RATING_PREFIX:")[1].trim();
            System.out.println(">>>>>>>target rating data is >>>>>> " + input);
            processorContext.forward("logProcessor".getBytes(), input.getBytes());
        }
    }

    @Override
    public void punctuate(long timestamp) {
    }

    @Override
    public void close() {
    }
}
